package com.example.demo;

/**
 * @Author lizhien (zhien.li@nascent.cn)
 * @Date 2020/8/4 19:51
 * @Version 1.0.0
 * @Description com.example.demo
 */

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.Session;
import javax.jms.TextMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ScheduledMessage;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

@RestController
@Slf4j
public class ClearScheduleMessageController {

    //原MQ地址
    private String brokerurl = "nio://localhost:61616"; //设置发布地址为默认的URL地址
    private Connection connection;
    private MessageProducer producer;
    private Session session;

    //新MQ地址
    private String brokerurl1 = "nio://192.168.1.230:61616"; //设置发布地址为默认的URL地址
    private Connection connection1;

    //构造对象
    public ClearScheduleMessageController() throws JMSException {
        this.brokerurl = System.getenv("ACTIVEMQ_URL") != null && !System.getenv("ACTIVEMQ_URL").isEmpty() ? System.getenv("ACTIVEMQ_URL") : this.brokerurl;
        this.connection = new ActiveMQConnectionFactory("failover:(" + this.brokerurl + ")?randomize=true").createConnection(); //创建从工厂连接中得到的对象
        this.connection.start();

        this.brokerurl1 = System.getenv("ACTIVEMQ_URL1") != null && !System.getenv("ACTIVEMQ_URL1").isEmpty() ? System.getenv("ACTIVEMQ_URL1") : this.brokerurl1;
        this.connection1 = new ActiveMQConnectionFactory("failover:(" + this.brokerurl1 + ")?randomize=true").createConnection(); //创建从工厂连接中得到的对象
        this.connection1.start();
    }

    //发送消息
    @GetMapping("/sendMsg")
    @ResponseBody
    public R sendMsg(@RequestParam(value = "dest", defaultValue = "test") String dest, @RequestParam(value = "msg", defaultValue = "test") String msg, @RequestParam(value = "delay", defaultValue = "0") long delay) throws JMSException {
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //fasle：参数表示为非事务类型；AUTO_ACNOWLEDGE：消息自动确认
        producer = session.createProducer(session.createQueue(dest));
        producer.setDeliveryMode(DeliveryMode.PERSISTENT); //设置消息非持久化（学习用法）
        TextMessage message = session.createTextMessage(msg);
        if (delay > 0) {
            message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay); //设置延迟时间
        }
        producer.send(message);
        log.info("发送队列消息：" + dest + " 延迟：" + delay + " 内容：" + msg);
        return R.ok(new HashMap<String, String>() {
            {
                put("dest", dest);
                put("msg", msg);
                put("delay", String.valueOf(delay));
            }
        }, "sendMsg success");
    }

    //按时间戳清理
    @GetMapping("/clearByTime")
    public R clearByTime(@RequestParam("time") String time) {
        try {

            long start = System.currentTimeMillis() - TimeUnit.HOURS.toMillis(24000); //删除：当前时间前12小时范围的延时消息
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(time));
            long end = calendar.getTimeInMillis();

            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //fasle：参数表示为非事务类型；AUTO_ACNOWLEDGE：消息自动确认
            producer = session.createProducer(session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION));

            Message request = session.createMessage();
            request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVEALL);
            request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION_START_TIME, Long.toString(start));
            request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION_END_TIME, Long.toString(end));

            producer.send(request);
            log.info("清理过期定时消息时间：" + time);

        } catch (JMSException | ParseException e) {
            e.printStackTrace();
        }
        return R.ok("clearByTime success");
    }

    //按消息ID清理
    @GetMapping("/clearById")
    public R clearById(@RequestParam("id") String id) {
        try {
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //fasle：参数表示为非事务类型；AUTO_ACNOWLEDGE：消息自动确认
            producer = session.createProducer(session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION));

            Message request = session.createMessage();
            request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVE);
            request.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_ID, id);

            producer.send(request);
            log.info("删除定时消息：" + id);

        } catch (JMSException e) {
            e.printStackTrace();
        }
        return R.ok("clearById success");
    }

    //清理全部队列
    @GetMapping("/clearAllQueues")
    public R clearAllQueues(@RequestParam("type") int type) throws JMSException {

        //判断新老MQ连接
        ActiveMQConnection conn = (type == 1 ? (ActiveMQConnection) connection : (ActiveMQConnection) connection1);

        Set<ActiveMQQueue> queues = conn.getDestinationSource().getQueues();
        for (ActiveMQQueue queue : queues) {
            log.info("queue: " + queue);
            conn.destroyDestination(queue);
        }

        log.info("清理全部消息队列完成");
        return R.ok("clearAllQueues success");
    }

    //清理全部定时消息
    @GetMapping("/clearAllScheduled")
    public R clearAllScheduled(@RequestParam("type") int type) {
        try {
            if (type == 1) {
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //fasle：参数表示为非事务类型；AUTO_ACNOWLEDGE：消息自动确认
            } else if (type == 2) {
                session = connection1.createSession(false, Session.AUTO_ACKNOWLEDGE);
            }

            Message request = session.createMessage();
            request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVEALL);
            producer = session.createProducer(session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION));
            producer.send(request);
            log.info("清理全部定时消息");

        } catch (JMSException e) {
            e.printStackTrace();
        }
        return R.ok("clearAllScheduled success");
    }

    //移动全部定时消息
    @GetMapping("/moveAll")
    public R moveAll() throws JMSException {
        this.clearAllScheduled(2);
        this.clearAllQueues(2);
        return R.ok(this.moveAllScheduled(true, false, true).getData());
    }

    //手动发送过期定时消息
    @GetMapping("/resendAll")
    public R resendAll() throws JMSException {
        return R.ok(this.moveAllScheduled(true, false, false).getData());
    }

    //移动全部定时消息
    @GetMapping("/moveAllScheduled")
    public R moveAllScheduled(@RequestParam(value = "send", defaultValue = "true") boolean send,
        @RequestParam(value = "clear", defaultValue = "false") boolean clear,
        @RequestParam(value = "move", defaultValue = "false") boolean move) throws JMSException {
        //创建管理会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination requestDest = session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION);
        MessageProducer producer = session.createProducer(requestDest);

        //将queue中的消息全部拿出来放到replyDesc中浏览
        Destination replyDest = session.createTemporaryQueue();
        TextMessage request = session.createTextMessage();
        request.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_BROWSE);
        request.setJMSReplyTo(replyDest);
        producer.send(request);

        //使用枚举来遍历所有消息
        QueueBrowser requestBrowser = session.createBrowser((Queue) replyDest);
        Enumeration<ActiveMQTextMessage> enumer = requestBrowser.getEnumeration();
        int i = 0;
        List<String> resultSend = new ArrayList<>();
        List<String> resultMove = new ArrayList<>();
        while (enumer.hasMoreElements()) {
            i++;
            ActiveMQTextMessage curMsg = enumer.nextElement();

            //拿到消息的ID
            String scheduleID = curMsg.getStringProperty(ScheduledMessage.AMQ_SCHEDULED_ID);
            String text = curMsg.getText();
            long time = curMsg.getJMSTimestamp();
            long delay = curMsg.getLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY);
            ActiveMQDestination dest = curMsg.getOriginalDestination();

            long now = System.currentTimeMillis();
            //手动投递过期消息
            if (send && now >= time + delay) {
                String txt = "过期消息：" + i + " 队列：" + dest + " 时间：" + getFormatTime(time) + " 延时：" + delay + " 执行时间：" + getFormatTime(now + delay) + " 内容：" + text;
                log.info(txt);
                sendMsg(dest.getPhysicalName(), text, 0);
                //删除被移消息
                clearById(scheduleID);
                resultSend.add(txt);
            } else if (move && now < time + delay) {
                String txt = "未过期消息：" + i + " 队列：" + dest + " 时间：" + getFormatTime(time) + " 延时：" + delay + " 执行时间：" + getFormatTime(now + delay) + " 内容：" + text;
                log.info(txt);

                //创建新的MQ延迟消息
                Session session1 = connection1.createSession(false, Session.AUTO_ACKNOWLEDGE); //fasle：参数表示为非事务类型；AUTO_ACNOWLEDGE：消息自动确认
                MessageProducer producer1 = session1.createProducer(dest);
                producer1.setDeliveryMode(DeliveryMode.PERSISTENT);
                TextMessage message = session1.createTextMessage(text);
                message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, getDelayTime(time, delay)); //设置延迟时间

                //发送新消息
                producer1.send(message);
                //删除被移消息
                if (clear) {
                    clearById(scheduleID);
                }
                resultMove.add(txt);
            }
        }
        return R.ok(new HashMap<String, List>() {
            {
                put("send", resultSend);
                put("move", resultMove);
            }
        });
    }

    //计算新的延时时间
    private static long getDelayTime(long time, long delay) {
        long now = System.currentTimeMillis();
        return now >= time + delay ? 1L : delay - (now - time);
    }

    //转换时间格式
    private static String getFormatTime(long time) {
        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(time);
    }

}
